package se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.flp2p.Flp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;
import se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast.event.PbBroadcast;
import se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast.event.PbDeliver;
import se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast.messages.DataMessage;
import se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast.messages.RequestMessage;
import se.sics.kompics.tutorial.tutorial2.lazyProbabilisticBcast.timeout.LpbTimeout;
import se.sics.kompics.tutorial.tutorial2.unreliableBroadcast.UnreliableBroadcast;
import se.sics.kompics.tutorial.tutorial2.unreliableBroadcast.events.UnBroadcast;
import se.sics.kompics.tutorial.tutorial2.unreliableBroadcast.events.UnDeliver;

public final class LazyPBImpl extends ComponentDefinition {

	Positive<UnreliableBroadcast> unBroadcast = requires(UnreliableBroadcast.class);
	Positive<Timer> timer = requires(Timer.class);
	Positive<FairLossPointToPointLink> flp2p = requires(FairLossPointToPointLink.class);
	Negative<ProbabilisticBroadcast> pbBroadcast = provides(ProbabilisticBroadcast.class);
	
	private static final Logger logger = LoggerFactory.getLogger(LazyPBImpl.class);
	
	private static final Random RANDOM_GENERATOR = new Random(System.currentTimeMillis());
	
	// Saving rate of broadcasted messages
	private static final double STORE_THRESHOLD = 0.70;
	
	// Timeout on gossiped request
	private static final long DELTA = 5000;
	
	// Number of node to contact for gossiping
	private static final int FANOUT = 1;
	
	// Max TTL of a gossip message
	private static final int MAX_TTL = 2;
	
	private String[] commands;
	private int lastCommand;
	private Set<Address> neighborSet;
	private Address self;
	private Integer counter;
	private Set<DataMessage> pending = new HashSet<DataMessage>(), stored = new HashSet<DataMessage>();
	private Map<Address, Integer> nextDelivery;
	
	public LazyPBImpl(){
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handlePbBcastRequest, pbBroadcast);
		subscribe(handleUnBcastedReception, unBroadcast);
		subscribe(handleRequestMessage, flp2p);
		subscribe(handleDataMessage, flp2p);
		subscribe(handleLpbTimeout, timer);
	}
	
	
	// Init Handler
	Handler<PbreliableBroadcastInit> handleInit = new Handler<PbreliableBroadcastInit>() {

		@Override
		public void handle(PbreliableBroadcastInit initEvent) {
			// Topology init
			neighborSet = initEvent.getNeighborSet();
			self = initEvent.getSelf();
			commands = initEvent.getCommandScript().split(":");
			lastCommand = -1;
			
			// State init
			counter = 0;
			nextDelivery = new HashMap<Address, Integer>();
			for(Address neighbor : neighborSet){
				nextDelivery.put(neighbor, 1);
			}
		}
	};
	
	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
			doNextCommand();
		}
	};
	
	// Broadcast a message
	Handler<PbBroadcast> handlePbBcastRequest = new Handler<PbBroadcast>() {

		@Override
		public void handle(PbBroadcast event) {
			counter ++;
			trigger(new UnBroadcast(new DataMessage(self, counter, event.getMessage())), unBroadcast);
			logger.info("Sending message: {}", event.getMessage());
		}
		
	};
	
	// Receive a broadcasted message
	Handler<UnDeliver> handleUnBcastedReception = new Handler<UnDeliver>() {
		
		@Override
		public void handle(UnDeliver event) {
			DataMessage message = event.getDataMessage();
			
			// Save the message with proba 1-STORE_THRESHOLD
			if(RANDOM_GENERATOR.nextDouble() > STORE_THRESHOLD){
				stored.add(message);
				logger.info("Stored: {}", message.getMessage());
			}
			
			// If there is no gap in the messages from this source
			if(message.getCounter() == nextDelivery.get(message.getSource())){
				nextDelivery.put(message.getSource(), message.getCounter()+1);
				trigger(new PbDeliver(
							message.getSource(), 
							message.getMessage())
						, pbBroadcast);
				doDeliveryPending(message.getSource());
			}
			// Else, if we are missing some messages
			else if(message.getCounter() > nextDelivery.get(message.getSource())){
				pending.add(message);
				// For all missing messages
				for(int i = nextDelivery.get(message.getSource()); i < message.getCounter(); i++){
					// If the message is not in the pending messages
					if(doStoreLookup(message.getSource(), i, pending) == null){
						// Gossip a request to retrieve it
						doGossip(new RequestMessage(self, i, message.getSource(), MAX_TTL));
						// Schedule the timeout
						ScheduleTimeout requestTimeout = new ScheduleTimeout(DELTA);
						requestTimeout.setTimeoutEvent(new LpbTimeout(requestTimeout, message.getSource(), message.getCounter()));
						trigger(requestTimeout, timer);
					}
				}
			}
		}
	};
	
	// Handle gossipes request messages
	Handler<RequestMessage> handleRequestMessage = new Handler<RequestMessage>() {

		@Override
		public void handle(RequestMessage event) {
			logger.info("Request received from {}.", event.getSource());
			// If we have the relevant message saved
			DataMessage message = doStoreLookup(event.getRequestedSource(), event.getCounter(), stored);
			if(message != null){
				logger.info("I have the message, I'm sending it.");
				trigger(new Flp2pSend(event.getSource(), message), flp2p);
			}
			// Else, regossip the request, if its ttl was not null
			else {
				if(event.getTtl() > 1){
					logger.info("I don't have the message, I'm forwarding the request");
					RequestMessage forward = new RequestMessage(event.getSource(),
																event.getCounter(),
																event.getRequestedSource(), 
																event.getTtl()-1);
					doGossip(forward);
				} else {
					logger.info("TTL of the request has reached 0, not forwarding the request.");
				}
			}
		}
		
	};
	
	// Handle the response to a gossiped request
	Handler<DataMessage> handleDataMessage = new Handler<DataMessage>() {

		@Override
		public void handle(DataMessage event) {
			logger.info("Message received: {}", event.getMessage());
			if (event.getCounter() == nextDelivery.get(event.getSource())) {
				nextDelivery.put(event.getSource(), event.getCounter() + 1);
				trigger(new PbDeliver(
						event.getSource(), 
						event.getMessage())
					, pbBroadcast);
				doDeliveryPending(event.getSource());
			} else {
				pending.add(event);
			}
		}
		
	};
	
	// Handle the timeout for gossiped request
	Handler<LpbTimeout> handleLpbTimeout = new Handler<LpbTimeout>() {

		@Override
		public void handle(LpbTimeout event) {
			if(event.getCounter() > nextDelivery.get(event.getSender())){
				nextDelivery.put(event.getSender(), event.getCounter()+1);
			}
		}
		
	};
	
	
	// Return the counter-th message from source in the corresponding store or null
	private DataMessage doStoreLookup(Address source, int counter, Set<DataMessage> store) {
		for(DataMessage message : store){
			if(message.getCounter() == counter && message.getSource().equals(source)){
				return message;
			}
		}
		return null;
	}

	// Gossip a request to FANOUT different nodes
	private void doGossip(RequestMessage requestMessage) {
		logger.info("Gossiping request for lsn {}, by {}.",
				requestMessage.getCounter(), requestMessage.getRequestedSource());
		Set<Address> toSend = new HashSet<Address>();
		int size = neighborSet.size();
		while(toSend.size() < FANOUT) {
			int item = RANDOM_GENERATOR.nextInt(size);
			int i = 0;
			for(Address neighbor : neighborSet)
			{
				if (i == item)
					toSend.add(neighbor);
				i = i + 1;
			}
		}
		for(Address neighbor : toSend) {
			trigger(new Flp2pSend(neighbor, requestMessage), 
					flp2p);
		}
	}
	
	private void doDeliveryPending(Address neighbor) {
		while(pending.iterator().hasNext()) {
			DataMessage data = pending.iterator().next();
			if (data.getSource().equals(neighbor) && data.getCounter() == nextDelivery.get(neighbor)) {
				nextDelivery.put(neighbor, nextDelivery.get(neighbor) + 1);
				pending.remove(data);
				trigger(new PbDeliver(data.getSource(), data.getMessage()), pbBroadcast);
				doDeliveryPending(neighbor);
			}
		}
	}
	
	private final void doNextCommand() {
		lastCommand++;

		if (lastCommand > commands.length) {
			return;
		}
		if (lastCommand == commands.length) {
			logger.info("DONE ALL OPERATIONS");
			Thread applicationThread = new Thread("ApplicationThread") {
				public void run() {
					BufferedReader in = new BufferedReader(
							new InputStreamReader(System.in));
					while (true) {
						try {
							String line = in.readLine();
							doCommand(line);
						} catch (Throwable e) {
							e.printStackTrace();
						}
					}
				}
			};
			applicationThread.start();
			return;
		}
		String op = commands[lastCommand];
		doCommand(op);
	}

	private void doCommand(String cmd) {
		if (cmd.startsWith("B")) {
			handlePbBcastRequest.handle(new PbBroadcast(cmd.substring(1)));
			doNextCommand();
		} else if (cmd.equals("help")) {
			doHelp();
			doNextCommand();
		} else if (cmd.startsWith("X")) {
			doShutdown();
		} else {
			logger.info("Bad command: '{}'. Try 'help'", cmd);
			doNextCommand();
		}
	}

	private final void doHelp() {
		logger.info("Available commands: B<m>, help, X");
		logger.info("Bm: broadcast message 'm' to all neighbors");
		logger.info("help: shows this help message");
		logger.info("X: terminates this process");
	}
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
